package com.apps.sdses.flink141.cdc;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2024/1/11 15:03
 */
public class TestJon {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //source
        tableEnv.executeSql("CREATE TABLE table_1 (\n" +
                "    id INT NOT NULL,\n" +
                "    s_id STRING,\n" +
                "    suggest_heat_2 DECIMAL(10, 2),\n" +
                "    create_time TIMESTAMP(3),\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'server-time-zone' = '+08:00',\n" +
                "    'database-name' = 'mydb',\n" +
                "    'server-id' = '5401-5405',\n" +
                "    'table-name' = 'table_1'\n" +
                "  )");

        tableEnv.executeSql("CREATE TABLE table_2 (\n" +
                "    id INT NOT NULL,\n" +
                "    s_id STRING,\n" +
                "    v1 DECIMAL(10, 2)\n" +
                "  ) WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'drity_data_source',\n" +
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'json.ignore-parse-errors' = 'true',\n" +
                "  'format' = 'json'\n" +
                "  )");

        tableEnv.executeSql("create TEMPORARY VIEW view_1 as select \n" +
                "\tt1.s_id,\n" +
                "\tt2.suggest_heat_2 \n" +
                "from (select s_id, max(create_time) as ct from table_1 t group by s_id) as t1\n" +
                "left join table_1 as t2\n" +
                "on t1.ct = t2.create_time and t1.s_id = t2.s_id ");


        //sink
        tableEnv.executeSql("CREATE TABLE sinkTable(s_id STRING, v1 DECIMAL(10,2), suggest_heat_2 DECIMAL(10, 2)) WITH ('connector' = 'print')");


        tableEnv.executeSql("insert into sinkTable " +
                "select t.s_id, t.v1, v.suggest_heat_2 " +
                "from table_2 as t left join view_1 as v on t.s_id = v.s_id");


    }
}
